package kafka.mq.producer;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;

import kafka.javaapi.producer.Producer;
import kafka.mq.entity.ResumePost;
import kafka.mq.util.JsonUtils;
import kafka.mq.util.SMS;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bj58.spat.esbclient.ESBClient;
import com.bj58.spat.esbclient.ESBMessage;
import com.bj58.spat.esbclient.ESBReceiveHandler;
import com.bj58.spat.esbclient.ESBSubject;

//async方式单Producer
/**
 * 从公司的消息总线里接收消息，发送到kafka，供spark消费
 */
public class FurtherMqProducer {
	private static final Logger logger = LoggerFactory.getLogger(FurtherMqProducer.class);
	private Producer<String,String> producer;
	private AtomicLong counter = new AtomicLong(0); 
	private ESBClient client;
	private String TOPIC;
	private int numOfPartitions;
	
	private FurtherMqProducer(){
		initProducer();
		registerExitEvent();
	}
	
	private static class SingletonHolder {  
		private static final FurtherMqProducer INSTANCE = new FurtherMqProducer();  
	} 
	
	public static final FurtherMqProducer getInstance() {  
		 return SingletonHolder.INSTANCE;  
	} 
	
	/*
	 * 加载配置文件，初始化Producer
	 */
	private  void initProducer() {
		try {
			InputStream inStream = FurtherMqProducer.class.getResourceAsStream("/kafka-producer-conf.properties");
			Properties props = new Properties();
			props.load(inStream);
			this.TOPIC = props.getProperty("topic");
			this.numOfPartitions = Integer.valueOf(props.getProperty("partition.num"));
			//这两个参数是自己加的，不属于kafka的参数
			props.remove("topic");
			props.remove("partition.num");
			ProducerConfig config = new ProducerConfig(props);
			producer = new Producer<String,String>(config);
		} catch (IOException e) {
			logger.error("ERROR Occurs when init the IMCProducer: \n",e);
			System.exit(1);
		}
	}

	/*
	 * 添加回调方法，关闭producer
	 */
    private  void registerExitEvent()
    {
        Runtime.getRuntime().addShutdownHook(new Thread() {
	            public void run() {
	            	if(producer !=null){
	            		producer.close();
	            	}
	            	if(client != null){
	            		client.closeAllChannel();
	            	}
	            }
        	}
        );
    }
    
	/*
	 * 加载配置文件，初始化 ESBClient
	 */
	private  void initESBClient() {
		try {
			InputStream inStream = FurtherMqProducer.class.getResourceAsStream("/esb-conf.properties");
			Properties props = new Properties();
			props.load(inStream);
			inStream.close();
			client = new ESBClient(props.getProperty("url"));
			client.setReceiveSubject(new ESBSubject(Integer.parseInt(props.getProperty("subjectId"))));
			client.setReceiveHandler(new ESBReceiveHandler() {
				@Override
				public void messageReceived(ESBMessage msg) {
		            String body = new String(msg.getBody());
		            ResumePost resumePost = JsonUtils.fromJSON(body, ResumePost.class);
		            if(resumePost !=null && resumePost.getOper() == 1 && StringUtils.equalsIgnoreCase("postResume", resumePost.getWhichMethod())){
		            	long num = counter.getAndIncrement();
			            KeyedMessage<String, String> data = new KeyedMessage<String, String>(TOPIC,String.valueOf(num % numOfPartitions), resumePost.getEntity());
			            producer.send(data);
		            }
		            logger.info(body);
				}
			});
			// 保持监听状态，让ESB主线程不消失
			while (true) {
				Thread.sleep(1000 * 60 * 60);
			}
		} catch (Exception e) {
			SMS.defaultSend("IMC-Kafka异常退出!请及时查看重启!");
			logger.error("ERROR Occurs : ",e);
		}finally{
			System.exit(1);
		}
	}
		
	protected void start() {
		try {
			if(StringUtils.isBlank(TOPIC)){
				logger.error("TOPIC is null.");
				System.exit(1);
			}
			
			initESBClient();
			
		} catch (Exception e) {
			logger.error("ERROR Occurs in function start() program will exit: ",e);
		}finally{
			System.exit(1);
		}
    }
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		FurtherMqProducer producer = FurtherMqProducer.getInstance();
		producer.start();
	}
}
